/*
 * Copyright (c) Huawei Technologies Co., Ltd. 2022-2022. All rights reserved.
 */

import { noop } from '../../../../common/util/noop.js';
import { CoreTask } from '../../../external/task/core-task.js';
import { CoreTaskImpl } from '../../../external/task/core-task-impl.js';
import { HvigorLogger } from '../../../log/hvigor-log.js';
import { DurationEvent, DurationEventState } from '../../../metrics/event/duration-event.js';
import { MetricLogType } from '../../../metrics/event/log-event.js';
import { MetricService } from '../../../metrics/metric-service.js';
import { endMonitor, startMonitor } from '../../../util/cpu.js';
import { coreParameter } from '../../data/global-core-parameters.js';
import { PoolConstant } from '../constant/constant.js';
import { Dispatcher } from '../dispatcher/dispatcher.js';
import { DispatcherFactory } from '../dispatcher/dispatcher-factory.js';
import { DispatcherType } from '../enum/dispatcher-type.js';
import { LogType } from '../enum/log-type.js';
import { PoolState, transitionMap } from '../enum/pool-state.js';
import { Priority } from '../enum/priority.js';
import { Log, WorkLog } from '../log-collection/log.js';
import { LogService } from '../log-collection/log-service.js';
import { LogServiceImpl } from '../log-collection/log-service-impl.js';
import { ITCB } from '../model/itcb.js';
import { TCB } from '../model/tcb.js';
import { Work } from '../model/work.js';
import { ClusterPriorityReadyQueue } from '../ready-queue/cluster-priority-ready-queue.js';
import { ReadyQueue } from '../ready-queue/ready-queue.js';
import { TcbStore } from '../store/tcb-store.js';
import { WorkerManager } from '../worker-manager/worker-manager.js';
import { WorkerManagerImpl } from '../worker-manager/worker-manager-impl.js';

import { isSamePoolConfig, PoolConfig } from './pool-config.js';
import { SubmitOption } from './submit-option.js';
import { WorkerPool } from './worker-pool.js';

const log = HvigorLogger.getLogger('hvigor-worker-pool');

/**
 * 工作池的实现类
 *
 * @since 2022/8/12
 */
export class WorkerPoolImpl implements WorkerPool {
  private static instance: WorkerPoolImpl | undefined;
  private readonly readyQueue: ReadyQueue;
  private dispatcher: Dispatcher;
  private readonly workerManager: WorkerManager;
  private readonly logService: LogService;
  private state: PoolState;
  private resident: boolean; // 一次构建结束后是否常驻
  private readonly poolConfig: PoolConfig;

  private constructor(poolConfig: PoolConfig) {
    log.debug('Worker pool is initialized with config: ', poolConfig);
    this.poolConfig = poolConfig;
    this.resident = true; // 默认为true
    this.state = PoolState.INIT;
    this.dispatcher = DispatcherFactory.getDispatcher(DispatcherType.LOOP_PRIORITY)!;
    this.logService = new LogServiceImpl();
    (this.logService as LogServiceImpl).start();
    this.workerManager = new WorkerManagerImpl(
      poolConfig,
      (workerManager: WorkerManager, workerId?: number) => {
        return this.dispatcher.dispatch(this.logService as LogServiceImpl, this.readyQueue, workerManager, workerId);
      },
      (log: WorkLog) => {
        (this.logService as LogServiceImpl).addLog(log);
      },
    );
    this.readyQueue = new ClusterPriorityReadyQueue(this.workerManager.getMaxPoolNum());
    startMonitor();
    this.state = PoolState.RUNNING;
    if (!this.checkPoolState()) {
      this.terminate();
    }
  }

  setResident(resident: boolean): void {
    this.resident = resident;
  }

  isResident(): boolean {
    return this.resident;
  }

  getMaxPoolNum(): number {
    return this.workerManager.getMaxPoolNum();
  }

  getMinPoolNum(): number {
    return this.workerManager.getMinPoolNum();
  }

  /**
   * 获取工作池的实例
   *
   * @param poolConfig 池配置
   * @returns {WorkerPool} 工作池实例
   */
  static getInstance(poolConfig: PoolConfig): WorkerPool {
    if (!WorkerPoolImpl.instance || WorkerPoolImpl.instance.state === PoolState.TERMINATED) {
      log.debug('Since there is no instance or instance is terminated, create a new worker pool.');
      WorkerPoolImpl.instance = new WorkerPoolImpl(poolConfig);
    } else {
      if (!isSamePoolConfig(WorkerPoolImpl.instance.getPoolConfig(), poolConfig)) {
        log.debug('Current worker pool does not match this config, create a new worker pool.');
        log.debug('Current pool config: ', WorkerPoolImpl.instance.poolConfig);
        log.debug('New pool config: ', poolConfig);
        WorkerPoolImpl.instance.terminate().then((res) => log.debug(`worker pool terminated with: ${res}`));
        WorkerPoolImpl.instance = new WorkerPoolImpl(poolConfig);
      }
    }
    return WorkerPoolImpl.instance!;
  }

  getLog(type: LogType): Set<Log> | undefined {
    return this.logService.getLog(type);
  }

  clearLog(type?: LogType): void {
    return this.logService.clear(type);
  }

  getState(): PoolState {
    return this.state;
  }

  setMaxIdleTime(maxIdleTime: number): void {
    this.workerManager.setMaxIdleTime(maxIdleTime);
  }

  setRecycleInterval(recycleInterval: number): void {
    this.workerManager.setRecycleInterval(recycleInterval);
  }

  setErrorCallback(errorCallback: Function): void {
    this.workerManager.setErrorCallback(errorCallback);
  }

  setState(state: PoolState): void {
    if (transitionMap.get(this.state)!.has(state)) {
      this.state = state;
      if (this.state === PoolState.STOP) {
        this.readyQueue.clear();
      }
    }
  }

  warmUp(workPath: string): boolean {
    if (this.state !== PoolState.RUNNING) {
      return false;
    }
    const work = new Work(Priority.FIRST, undefined, workPath);
    const tcb: TCB = new ITCB(
      work,
      noop,
      false,
      [],
      () => {
        this.readyQueue.update(work);
      },
      '',
      '',
      '',
    );
    TcbStore.add(tcb);
    this.readyQueue.push(work);
    this.dispatcher.dispatch(this.logService as LogServiceImpl, this.readyQueue, this.workerManager);
    return true;
  }

  // 考虑如何通过注入或者其他方式，而不通过传参的方式传递该task
  submit(task: CoreTask, workPath: string, submitOption?: SubmitOption): TCB {
    const work = new Work(
      submitOption?.priority ?? Priority.MEDIUM,
      this.filterTargetWorkers(submitOption?.targetWorkers),
      workPath,
      submitOption?.workInput,
      submitOption?.hasSideEffects,
      submitOption?.preludeDeps,
      submitOption?.memorySensitive,
    );
    return this.submitWork(task, work, submitOption);
  }

  /**
   * 过滤出合法的worker集合
   *
   * @param targetWorkers 待过滤的worker集合
   * @returns {number[]} 过滤结果
   */
  private filterTargetWorkers(targetWorkers: number[] | undefined): number[] | undefined {
    if (!targetWorkers) {
      return targetWorkers;
    }
    const workerMap: Set<number> = new Set();
    const res: number[] = [];
    for (const worker of targetWorkers) {
      if (Number.isInteger(worker) && !workerMap.has(worker) && worker < this.workerManager.getMaxPoolNum() && worker >= 0) {
        workerMap.add(worker);
        res.push(worker);
      }
    }
    return res;
  }

  private submitWork(task: CoreTask, work: Work, submitOption?: SubmitOption): TCB {
    if (!this.workSubmissionCanBeAccepted(work)) {
      log.debug(`Task ${task.getPath()}'s work:${work.getId()} will not be executed in worker pool.`);
      return ITCB.getDefaultRejectTcb(task, work, submitOption);
    }
    const taskInfo = `${task.getName()} work[${work.getId()}]`;
    log.debug(`${taskInfo} is submitted.`);
    const tcb: TCB = this.initTcb(task, work, submitOption);
    this.readyQueue.push(work);
    log.debug(`${taskInfo} is pushed to ready queue.`);
    if (this.dispatcher.dispatch(this.logService as LogServiceImpl, this.readyQueue, this.workerManager)) {
      log.debug(`${taskInfo} is dispatched.`);
    } else {
      log.debug(`${taskInfo} is not dispatched.`);
    }
    return tcb;
  }

  /**
   * 当用户提交work时，需要判断是否允许在WorkerPool中执行该work
   * 需要满足:
   * 1.用户开启并行能力，默认开启
   * 2.WorkerPool初始化完成
   * 3.指定的TargetWorker是符合要求的
   *
   * @param {Work} work
   * @return {boolean}
   * @private
   */
  private workSubmissionCanBeAccepted(work: Work): boolean {
    const isParallelExecution = coreParameter.startParams.parallelExecution;
    const isWorkerPoolReady = this.state === PoolState.RUNNING;
    const hasValidTargetWorker = work.getTargetWorkers() === undefined || work.getTargetWorkers()!.length !== 0;

    return isParallelExecution && isWorkerPoolReady && hasValidTargetWorker;
  }

  private initTcb(task: CoreTask, work: Work, submitOption?: SubmitOption): TCB {
    const tcb: TCB = new ITCB(
      work,
      submitOption?.callback ?? noop,
      submitOption?.useReturnVal,
      submitOption?.callbackInput ?? [],
      () => {
        this.readyQueue.update(work);
      },
      task.getNode().getName(),
      task.getName(),
      task.getPath(),
    );

    const coreTask = task as CoreTaskImpl;
    const promise = new Promise((resolve, reject) => {
      (tcb as ITCB).on(PoolConstant.WORK_DONE, (workId) => {
        log.debug(`${task.getName()} work[${workId}] done.`);
        coreTask.taskExecutedStatus.setWorkerTimePeriod([
          tcb.getStartTime() ?? Number(process.hrtime.bigint()),
          tcb.getEndTime() ?? Number(process.hrtime.bigint()),
        ]);
        this.recordDurationEvent(coreTask, workId, tcb, true);
        resolve(true);
      });
      (tcb as ITCB).on(PoolConstant.WORK_ERROR, (errorInfo: { id: string; error: Error }) => {
        log.debug(`${task.getName()} work[${errorInfo.id}] failed.`);
        this.recordDurationEvent(coreTask, errorInfo.id, tcb, false);
        reject(errorInfo.error);
      });
      (tcb as ITCB).on(PoolConstant.CALLBACK_ERROR, (errorInfo: { id: string; error: Error }) => {
        log.debug(`Callback of work[${errorInfo.id}] in ${task.getName()} failed.`);
        reject(errorInfo.error);
      });
    });
    coreTask.pendingPromises.add(promise);
    coreTask.addSubDurationEvent(work.getId());
    TcbStore.add(tcb);
    return tcb;
  }

  recordDurationEvent(coreTask: CoreTaskImpl, workId: string, tcb: TCB, result: boolean) {
    const subDurationEvent = coreTask.getSubDurationEvent(workId);
    subDurationEvent?.setTid(`Worker${tcb.getWorkerId()!}`);
    subDurationEvent?.start(DurationEventState.RUNNING, tcb.getStartTime());
    subDurationEvent?.stop(result ? DurationEventState.SUCCESS : DurationEventState.FAILED, tcb.getEndTime());
    subDurationEvent?.setLog(subDurationEvent?.getName(), result ? MetricLogType.INFO : MetricLogType.ERROR, subDurationEvent?.getDescription());
    this.updateTid(subDurationEvent);
  }

  private updateTid(parentEvent?: DurationEvent) {
    if (!parentEvent) {
      return;
    }
    parentEvent.getChildren().forEach((child) => {
      const childEvent = MetricService.getInstance().getEventById(child);
      this.updateTid(childEvent?.setTid(parentEvent.getTid()) as DurationEvent);
    });
  }

  terminate(): Promise<boolean> {
    return new Promise<boolean>((resolve, reject) => {
      if (this.state === PoolState.TERMINATED) {
        resolve(true);
      }
      if (this.state === PoolState.RUNNING) {
        this.setState(PoolState.STOP);
      }
      if (this.state === PoolState.STOP || this.state === PoolState.CLOSE) {
        log.debug('Current worker pool is stopped or closed.');
        (this.logService as LogServiceImpl).end();
        this.readyQueue.clear();
        this.workerManager.clear(() => {
          endMonitor().then(() => {
            this.setState(PoolState.TERMINATED);
            log.debug('Current worker pool is terminated.');
            resolve(true);
          });
        });
      } else {
        log.debug('Current worker pool has been terminated.');
        endMonitor().then(() => resolve(false));
      }
    });
  }

  recycle(): Promise<boolean> {
    return new Promise<boolean>((resolve, reject) => {
      this.readyQueue.clear();
      this.workerManager.cleanUp(() => resolve(true));
    });
  }

  /**
   * 检查初始化后池状态是否正常
   *
   * @returns {boolean} 检查结果
   */
  private checkPoolState(): boolean {
    if (this.getMaxPoolNum() < 1 || this.getMaxPoolNum() < this.getMinPoolNum()) {
      return false;
    }
    return true;
  }

  private getPoolConfig() {
    return this.poolConfig;
  }
}
